package cn.xuexiyuan.flinkstudy.window;

import cn.xuexiyuan.flinkstudy.entity.CartInfo;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @Description: 基于数量的滑动和滚动窗口 demo
 *
 *
 * @Author 左龙龙
 * @Date 21-3-25
 * @Version 1.0
 **/
public class WindowDemo02 {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<String> lines = env.socketTextStream("localhost", 6666);

        // 2.transformation
        SingleOutputStreamOperator<CartInfo> cartDS = lines.map(new MapFunction<String, CartInfo>() {
            @Override
            public CartInfo map(String s) throws Exception {
                String[] split = s.split(",");
                return new CartInfo(split[0], Integer.parseInt(split[1]));
            }
        });

        // 需求： 要求各个路口红绿灯的结果，所以需要分组
        KeyedStream<CartInfo, String> keyedDS = cartDS.keyBy(CartInfo::getSensorId);

        // 需求一： 统计最近5条消息中, 各自路口通过的汽车数量, 相同的key每出现5次进行统计 —— 基于数量的滚动窗口
        SingleOutputStreamOperator<CartInfo> result1 = keyedDS.countWindow(5).sum("count");

        // 需求二： 统计最近5条消息中, 各自路口通过的汽车数量, 相同的key每出现3次进行统计 —— 基于数量的滑动窗口
        SingleOutputStreamOperator<CartInfo> result2 = keyedDS.countWindow(5, 3).sum("count");

        // 3.sink
        result1.print("[Tumbling] ");
        result2.print("[Sliding] ");

        // 4.excute
        env.execute("WindowDemo02");
    }

}


